#include <stddef.h>
#include <stdint.h>
#include <string.h>
#include <limits.h>
#include <condition_variable>
#include <mutex>
#include <iostream>
#include <thread>

#include "raptor-lite/raptor-lite.h"

struct echo_message {
    uint32_t payload_size;
    uint32_t request_id;
};

bool connect_success = false;
std::mutex g_mtx;
std::condition_variable g_cv;
FILE *g_file = NULL;
int64_t g_file_size = 0;

class FileClient : public raptor::ConnectorHandler,
                   public raptor::MessageHandler,
                   public raptor::ProtocolHandler,
                   public raptor::EndpointClosedHandler {
public:
    FileClient(/* args */);
    ~FileClient();

    void OnConnectFailed(intptr_t user, raptor_error desc) {
        log_error("OnConnectFailed: %lld, %s", user, desc->ToString().c_str());
        connect_success = false;
        g_cv.notify_one();
    }

    void OnConnect(raptor::Endpoint ep, raptor::Property &settings) {
        settings({{"SocketRecvTimeoutMs", 5000}, {"SocketSendTimeoutMs", 5000}});

        raptor_error e = container->AttachEndpoint(ep);
        if (e != RAPTOR_ERROR_NONE) {
            log_error("AttachEndpoint:%s", e->ToString().c_str());
        }
        connection_id = ep.ConnectionId();
        log_info("AttachEndpoint: connection_id = %llu", connection_id);
        connect_success = true;
        g_cv.notify_one();
    }

    int OnCheckPackageLength(raptor::Endpoint ep, const void *data, size_t len) {
        // log_debug("OnCheckPackageLength:%u", len);
        (void)ep;
        if (len < 4) return 0;
        uint32_t pack_len = *reinterpret_cast<const uint32_t *>(data);
        return static_cast<int>(pack_len) + 8;
    }

    void OnMessage(raptor::Endpoint ep, const raptor::Slice &msg) {
        const echo_message *header = reinterpret_cast<const echo_message *>(msg.begin());
        log_debug("OnMessage: req=%lu", header->request_id);
        (void)ep;
        if (header->payload_size > 0) {
            fwrite(header + 1, 1, header->payload_size, g_file);
            g_file_size += header->payload_size;
        }
        if (header->request_id == 9527) {
            log_debug("Recv File Finished: %lld", g_file_size);
            fclose(g_file);
        }
    }

    void OnClosed(raptor::Endpoint ep, raptor_error desc) {
        if (desc == RAPTOR_ERROR_NONE) {
            log_debug("OnClosed: %s", ep.PeerString().c_str());
        } else {
            log_debug("OnClosed: %s, %s", ep.PeerString().c_str(), desc->ToString().c_str());
        }
    }

    void init() {

        // uint32_t cpus = raptor::GetNumberofCPUCores();
        engine = RaptorCreateEngine();
        if (engine == nullptr) {
            log_error("Failed to create engine");
            return;
        }

        raptor::Property p{
            {raptor::config::kConnectorHandler, static_cast<raptor::ConnectorHandler *>(this)},
            {raptor::config::kMessageHandler, static_cast<raptor::MessageHandler *>(this)},
            {raptor::config::kProtocolHandler, static_cast<raptor::ProtocolHandler *>(this)},
            {raptor::config::kEndpointClosedHandler,
             static_cast<raptor::EndpointClosedHandler *>(this)},
            {raptor::config::kCheckReceivingInterval, false},
            {raptor::config::kTcpUserTimeoutMs, 5000},
            {raptor::config::kRecvSendThreads, 1}};

        raptor_error err = engine->CreateContainer(p, &container);
        if (err != RAPTOR_ERROR_NONE) {
            log_error("Failed to create container: %s", err->ToString().c_str());
            abort();
        }
        err = engine->CreateConnector(p, &connector);
        if (err != RAPTOR_ERROR_NONE) {
            log_error("Failed to create connector: %s", err->ToString().c_str());
            abort();
        }
    }

    void start(const char *addr) {
        raptor_error err = connector->Start();
        if (err != RAPTOR_ERROR_NONE) {
            log_error("Failed to start connector: %s", err->ToString().c_str());
            return;
        }
        err = container->Start();
        if (err != RAPTOR_ERROR_NONE) {
            log_error("Failed to start container: %s", err->ToString().c_str());
            return;
        }
        err = connector->Connect(addr);
        if (err != RAPTOR_ERROR_NONE) {
            log_error("Failed to connect %s, error: %s", addr, err->ToString().c_str());
        }
    }

    void download(uint32_t request_id) {
        echo_message msg;
        msg.payload_size = 0;
        msg.request_id = request_id;
        if (connection_id != 0) {
            container->SendMsg(connection_id, raptor::Slice(&msg, sizeof(msg)));
        } else {
            log_error("connection not prepare");
        }
    }

    void stop() {
        connector->Shutdown();
        container->Shutdown();
    }

private:
    RaptorEngine engine;
    raptor::Container *container = nullptr;
    raptor::Connector *connector = nullptr;
    uint32_t connection_id = 0;
};

FileClient::FileClient(/* args */) {
    // Do not call the raptor::Create function here, because the current
    // class (inherited from XXHandler) has not been constructed.
}

FileClient::~FileClient() {
    engine->DestroyConnector(connector);
    engine->DestroyContainer(container);
}

int main() {
    printf(" ---- prepare start echo client ---- \n");
    RaptorGlobalStartup();
    FileClient client;
    client.init();
    client.start("localhost:50051");

    std::unique_lock<std::mutex> lck(g_mtx);
    g_cv.wait(lck);

    if (!connect_success) {
        std::this_thread::sleep_for(std::chrono::microseconds(5000));
        client.stop();
        RaptorGlobalCleanup();
        return 0;
    }

    log_warn("Press any key to quit");
    g_file = fopen("test.bin", "wb+");
    client.download(10086);
    getchar();
    client.stop();
    RaptorGlobalCleanup();

    return 0;
}
